#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/redirect_error.hpp>

#include "base/thread/thread_worker.hpp"
#include "rtmp_media_sink.hpp"
#include "rtmp_media_source.hpp"
#include "spdlog/spdlog.h"
using namespace mms;

RtmpMediaSink::RtmpMediaSink(ThreadWorker *worker) : MediaSink(worker)
{
}

bool RtmpMediaSink::init()
{
    return true;
}

RtmpMediaSink::~RtmpMediaSink()
{
    spdlog::info("destroy rtmp media sink");
}

bool RtmpMediaSink::on_audio_packet(std::shared_ptr<RtmpMessage> audio_pkt)
{
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self, audio_pkt]() -> boost::asio::awaitable<void>
                          {
        std::vector<std::shared_ptr<RtmpMessage>> v = {audio_pkt};
        co_await data_cb_(v);
        co_return; }, boost::asio::detached);
    return true;
}

bool RtmpMediaSink::on_video_packet(std::shared_ptr<RtmpMessage> video_pkt)
{
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self, video_pkt]() -> boost::asio::awaitable<void>
                          {
        std::vector<std::shared_ptr<RtmpMessage>> v = {video_pkt};
        co_await data_cb_(v);
        co_return; }, boost::asio::detached);
    return true;
}

bool RtmpMediaSink::on_metadata(std::shared_ptr<RtmpMessage> metadata_pkt)
{
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self, metadata_pkt]() -> boost::asio::awaitable<void>
                          {
        std::vector<std::shared_ptr<RtmpMessage>> v = {metadata_pkt};
        co_await data_cb_(v);
        co_return; }, boost::asio::detached);
    return true;
}

void RtmpMediaSink::wakeup()
{
    if (working_)
    {
        return;
    }
    working_ = true;
    auto self(this->shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(), [this, self]() -> boost::asio::awaitable<void>
        {
        if (!source_->is_ready()) {
            working_ = false;
            co_return;
        }
        auto rtmp_source = std::static_pointer_cast<RtmpMediaSource>(source_);
        std::vector<std::shared_ptr<RtmpMessage>> pkts;
        pkts.reserve(10);
        do {
            pkts = rtmp_source->get_pkts(last_send_pkt_index_, 10);
            if (!co_await data_cb_(pkts)) {
                break;
            }
        } while (pkts.size() > 0);

        working_ = false;
        co_return; }, boost::asio::detached);
}

void RtmpMediaSink::on_rtmp_message(const std::function<boost::asio::awaitable<bool>(const std::vector<std::shared_ptr<RtmpMessage>> &msgs)> &cb)
{
    data_cb_ = cb;
}

void RtmpMediaSink::close()
{
    data_cb_ = {};
    auto self(shared_from_this());
    boost::asio::co_spawn(worker_->get_io_context(),[this,self]()->boost::asio::awaitable<void>{
      spdlog::info(" source_->remove_media_sink(self);");
      source_->remove_media_sink(self);
      co_return;
    },boost::asio::detached);
    MediaSink::close();
}